package com.heima.schedule.service.impl;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.heima.common.cache.CacheService;
import com.heima.common.constants.ScheduleConstants;
import com.heima.model.schedule.dto.Task;
import com.heima.model.schedule.pojos.Taskinfo;
import com.heima.model.schedule.pojos.TaskinfoLogs;
import com.heima.schedule.service.TaskService;
import com.heima.schedule.service.TaskinfoLogsService;
import com.heima.schedule.service.TaskinfoService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;

import java.util.Date;
import java.util.List;
import java.util.Set;

/**
 * @author tongdulong@itcast.cn
 * @since 2022-07-13
 */
@Service
@Slf4j
@Transactional(rollbackFor = RuntimeException.class)
public class TaskServiceImpl implements TaskService {
	
	@Autowired
	private TaskinfoService taskinfoService;
	
	@Autowired
	private TaskinfoLogsService taskinfoLogsService;
	
	@Autowired
	private CacheService cacheService;
	
	@Override
	public long addTask(Task task) {
		// 添加任务到task
		long executeTime = task.getExecuteTime();
		
		Taskinfo taskInfo = new Taskinfo();
		BeanUtils.copyProperties(task, taskInfo);
		taskInfo.setExecuteTime(new Date(executeTime));
		
		boolean taskResult = taskinfoService.save(taskInfo);
		if (!taskResult) {
			log.warn("任务添加失败");
			throw new RuntimeException("任务添加失败");
		}
		
		// 添加任务到log
		TaskinfoLogs taskinfoLogs = new TaskinfoLogs();
		BeanUtils.copyProperties(task, log);
		taskinfoLogs.setExecuteTime(new Date(executeTime));
		taskinfoLogs.setStatus(ScheduleConstants.SCHEDULED);
		// taskinfoLogs.setVersion(1);
		boolean logResult = taskinfoLogsService.save(taskinfoLogs);
		if (!logResult) {
			log.warn("任务日志添加失败");
			throw new RuntimeException("任务日志添加失败");
		}
		
		addTaskToCache(task);
		
		// 返回任务id
		return taskInfo.getTaskId();
	}
	
	public void addTaskToCache(Task task) {
		long executeTime = task.getExecuteTime();
		long currentTimeMillis = System.currentTimeMillis();
		long futureTime = currentTimeMillis + (5 * 60 * 1000);
		// 如果小于当前就写入list
		String listKey = ScheduleConstants.TOPIC;
		// 如果小于五分钟后，就写入zset
		String zSetKey = ScheduleConstants.FUTURE;
		if (executeTime <= currentTimeMillis) {
			cacheService.lLeftPush(listKey, JSON.toJSONString(task));
		} else if (executeTime <= futureTime) {
			cacheService.zAdd(zSetKey, JSON.toJSONString(task), executeTime);
		}
	}
	
	@Transactional(rollbackFor = RuntimeException.class)
	public Boolean cancelTask(Long taskId) {
		if (taskId == null) {
			log.warn("taskId不能为空");
			return false;
		}
		//删除taskinfo
		boolean taskInfoResult = taskinfoService.removeById(taskId);
		if (!taskInfoResult) {
			log.warn("任务删除失败");
			return false;
		}
		
		//更新 Task结构
		LambdaQueryWrapper<TaskinfoLogs> wrapper = new LambdaQueryWrapper<>();
		wrapper.eq(TaskinfoLogs::getTaskId, taskId);
		
		TaskinfoLogs taskinfoLog = taskinfoLogsService.getOne(wrapper);
		
		taskinfoLog.setStatus(2);
		
		boolean logResult = taskinfoLogsService.updateById(taskinfoLog);
		if (!logResult) {
			log.warn("日志更新失败");
			throw new RuntimeException("日志更新失败");
		}
		
		//组装 Task结构
		
		Task task = new Task();
		BeanUtils.copyProperties(taskinfoLog, task);
		task.setExecuteTime(taskinfoLog.getExecuteTime().getTime());
		//删除redis 里面的 list 和 zset结构数据
		cacheService.lRemove("TOPIC", 0, JSON.toJSONString(task));
		cacheService.zRemove("FUTURE", JSON.toJSONString(task));
		
		return true;
	}
	
	//项目启动后自动定期运行.时间为毫秒.
	@Scheduled(fixedRate = 1000 * 1)
	public void refresh(){
		log.info("定时任务开始执行了");
		Set<String> future = cacheService.zRangeByScore("FUTURE", 0, System.currentTimeMillis());
		if (CollectionUtils.isEmpty(future)) {
			log.warn("没有可移动的任务");
			return;
		}
		cacheService.refreshWithPipeline("FUTURE", "TOPIC", future);
	}
	
	//清除缓存
	@Scheduled(fixedRate = 5 * 60 * 100)
	public void initData() {
		clear();
		
		List<Taskinfo> list = taskinfoService.list();
		
		if (CollectionUtils.isEmpty(list)) {
			log.warn("任务为空");
			return;
		}
		
		for (Taskinfo taskinfo : list) {
			if (taskinfo == null) {
				continue;
			}
			
			long executeTime = taskinfo.getExecuteTime().getTime();
			long currentTime = System.currentTimeMillis();
			long futrueTime = currentTime + (5 * 60 * 1000);
			
			Task task = new Task();
			BeanUtils.copyProperties(taskinfo,task);
			task.setExecuteTime(executeTime);
			
			if (executeTime <= currentTime) {
				cacheService.lLeftPush("TOPIC", JSON.toJSONString(task));
				
			} else if (executeTime <= futrueTime) {
				
				//第三个参数是当前任务的执行时间
				cacheService.zAdd("FUTURE", JSON.toJSONString(task), executeTime);
			}
			
		}
	}
	
	private void clear() {
		cacheService.delete("TOPIC");
		cacheService.delete("FUTURE");
	}
	
	@Override
	@Transactional(rollbackFor = RuntimeException.class)
	public Boolean poll() {
		// 1. 从topic弹出数据
		String taskString = cacheService.lRightPop("TOPIC");
		if (StringUtils.isEmpty(taskString)) {
			log.warn("没有可执行的任务");
			return false;
		}
		
		// 2. 装成task对象
		Task task = JSON.parseObject(taskString, Task.class);
		if (task == null) {
			log.warn("没有可执行的任务");
			return false;
		}
		
		boolean taskInfoResult = taskinfoService.removeById(task.getTaskId());
		if (!taskInfoResult) {
			log.warn("删除任务失败");
			return false;
		}
		
		long executeTime = task.getExecuteTime();
		
		TaskinfoLogs logs = new TaskinfoLogs();
		BeanUtils.copyProperties(task, logs);
		logs.setExecuteTime(new Date(executeTime));
		logs.setStatus(1);
		
		boolean logResult = taskinfoLogsService.updateById(logs);
		if (!logResult) {
			log.warn("日志更新失败");
			throw new RuntimeException("日志更新失败");
		}
		
		return true;
	}
}